package com.chencong.online.driver.dwd;

import com.chencong.online.bean.PageViewCountBean;
import com.chencong.online.bean.WebLogEventBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.HotPageTopNProcessFunc;
import com.chencong.online.function.PageCountAggFunc;
import com.chencong.online.function.PageCountWindowFunc;
import com.chencong.online.function.WebLogEventMapFunc;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.time.Duration;
import java.util.regex.Pattern;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdHotPages
 * @description:统计用户访问的热门页面
 * @author: chencong
 * @create: 2021-12-23 09:44
 **/
public class DwdHotPagesDriver {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        //1、source
        //通过反射拿到数据地址
        URL resource = DwdHotPagesDriver.class.getResource("/apache.log");
        DataStreamSource<String> inputDS = env.readTextFile(resource.getPath());
//        inputDS.print("初始化数据");
        //2、transform
        //将数据转换为WebLogEvent，过滤出想要字段和提取时间戳
        SingleOutputStreamOperator<WebLogEventBean> webLogEventDS = inputDS
                .map(new WebLogEventMapFunc())
                .filter(data -> "GET".equals(data.getMethod()))  //过滤出只是GET请求的访问url
                .filter(data -> {
                    String regex = "^((?!\\.(css|js|png|ico)$).)*$";//正则匹配规则
                    return Pattern.matches(regex, data.getUrl());
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.
                        <WebLogEventBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))  //设置最大乱序时间3s
                        .withTimestampAssigner(new SerializableTimestampAssigner<WebLogEventBean>() { //提取时间watermark的时间戳
                            @Override
                            public long extractTimestamp(WebLogEventBean element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        })
                );

//        webLogEventDS.print("转换后的数据");
        //开窗计算:滑动窗口，五分钟输出前一个小时的数据
        OutputTag<WebLogEventBean> lateDataTag = new OutputTag<WebLogEventBean>("late-date") {
        }; //定义迟到数据放入侧输出流
        SingleOutputStreamOperator<PageViewCountBean> pageViewCountAggDS = webLogEventDS
                .keyBy(WebLogEventBean::getUrl)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .allowedLateness(Time.minutes(1))  //运行迟到最大时间为1分钟
                .sideOutputLateData(lateDataTag)  //将迟到数据放入侧输出流
                .aggregate(new PageCountAggFunc(), new PageCountWindowFunc());

//        pageViewCountAggDS.print("每个窗口的统计数据为");

        DataStream<WebLogEventBean> lateDataSideOutput = pageViewCountAggDS.getSideOutput(lateDataTag); //从侧输出流获取迟到数据
//        lateDataSideOutput.print("迟到数据");

        //将同一窗口数据，开窗计算热门页面topN
        SingleOutputStreamOperator<String> resultDS = pageViewCountAggDS
                .keyBy(data -> data.getWindowEnd())
                .process(new HotPageTopNProcessFunc(5));
        resultDS.print("结果是：");
        env.execute("DwdHotPages");
    }
}
